package com.ydlclass.core;

import com.ydlclass.NettyBootstrapInitializer;
import com.ydlclass.Transport.message.Request.RequestPayload;
import com.ydlclass.Transport.message.Request.YrpcRequest;
import com.ydlclass.YrpcBootstrap;
import com.ydlclass.compress.CompressFactory;
import com.ydlclass.discovery.Registry;
import com.ydlclass.enumeration.RequestTypeEnum;
import com.ydlclass.serialize.SerializerFactory;
import io.netty.channel.Channel;
import io.netty.channel.ChannelFutureListener;
import lombok.extern.slf4j.Slf4j;

import java.net.InetSocketAddress;
import java.util.*;
import java.util.concurrent.*;

/**
 * 心跳检测器
 * 感知那些服务器的连接是正常的，哪些是不正常的
 *
 * @Author duWenJian
 * @Date 2024/2/26 16:30
 * @Version 1.0
 */
@Slf4j
public class HeartbeatDetector {
    /**
     * 检测服务是否存活
     * 将所有的可用服务，建立连接方便进行心跳检测
     * @param serviceName
     */
    public static void detectHeartbeat(String serviceName){
        log.info("开始进行心跳检测服务");
        // 1、遍历服务列表建立连接
        List<InetSocketAddress> socketAddressList = YrpcBootstrap.getInstance().getConfiguration().getRegistryConfig().getRegistry().lookup(serviceName,YrpcBootstrap.getInstance().getConfiguration().getGroup());
        // 2、将连接进行缓存
        for (InetSocketAddress address : socketAddressList) {
            try {
                if (!YrpcBootstrap.CHANNEL_CACHE.containsKey(address)){
                    Channel channel = NettyBootstrapInitializer.getBootstrap().connect(address).sync().channel();
                    // 将所有的连接channel都放置到缓存中
                    YrpcBootstrap.CHANNEL_CACHE.put(address,channel);
                }
            } catch (InterruptedException e) {
                throw new RuntimeException(e);
            }
        }

        log.info("心跳检测已经将所有的服务节点放置到缓存中");

        // 3、任务进行发送心跳检测
        Thread thread = new Thread(() -> {
            new Timer().scheduleAtFixedRate(new MyTimerTask(), 0, 2000);
        },"yrpc_HeartbeatDetector_thread");
        thread.setDaemon(true);
        thread.start();

    }
    /**
     * 定时任务
     */
    private static class MyTimerTask extends TimerTask {
        @Override
        public void run() {
            // 刚开始需要将响应时长的map清空
            YrpcBootstrap.ANSWER_TIME_CHANNEL_CACHE.clear();

            // 遍历所有的channel
            Map<InetSocketAddress, Channel> cache = YrpcBootstrap.CHANNEL_CACHE;

            for (Map.Entry<InetSocketAddress, Channel> entry : cache.entrySet()){
                int stampTimes = 3;
                while (stampTimes > 0 ){
                Channel channel = entry.getValue();
                long start = System.currentTimeMillis();
                YrpcRequest yrpcRequest = YrpcRequest.builder()
                        .requestId(YrpcBootstrap.getInstance().getConfiguration().getIdGenerator().getId())
                        .requestType(RequestTypeEnum.HeartbeatRequest.getId())
                        .compressType(CompressFactory.getCompress(YrpcBootstrap.getInstance().getConfiguration().getCompressionType()).getCode())
                        .serializeType(SerializerFactory.getSerializer(YrpcBootstrap.getInstance().getConfiguration().getSerializerType()).getCode())
                        .timeStamp(start)
                        .build();

                CompletableFuture<Object> future = new CompletableFuture<>();
                YrpcBootstrap.PENDING_REQUEST.put(yrpcRequest.getRequestId(),future);
                channel.writeAndFlush(yrpcRequest).addListener((ChannelFutureListener) promise -> {
                    if (!promise.isSuccess()) {
                        future.completeExceptionally(promise.cause());
                    }
                });

                long endTime = 0;
                try {
                    future.get(1, TimeUnit.SECONDS);
                    endTime = System.currentTimeMillis();
                } catch (InterruptedException | ExecutionException | TimeoutException e) {
                    stampTimes--;
                    log.error("和地址:{}的主机连接发生异常。正在进行第:{}次重试", channel.remoteAddress(),3-stampTimes);
                    // 将失效的地址移除, 并重新进行连接
                    if (stampTimes == 0){
                        YrpcBootstrap.CHANNEL_CACHE.remove(entry.getKey());
                        break;
                    }
                    // 等一段时间进行重试
                    try {
                        Thread.sleep(10*(new Random().nextInt(5)));
                    } catch (InterruptedException interruptedException) {
                        throw new RuntimeException(e);
                    }
                    continue;
                }
                Long time = endTime - start;

                // 使用treeMap进行缓存
                YrpcBootstrap.ANSWER_TIME_CHANNEL_CACHE.put(time,channel);
                log.debug("检测服务：{}，耗时：{}ms",entry.getKey(),time);
                break;
                }
            }


           log.info(".................................响应时间的TreeMap...............................");
            for (Map.Entry<Long,Channel> entry : YrpcBootstrap.ANSWER_TIME_CHANNEL_CACHE.entrySet()){
                log.info("响应时间："+entry.getKey()+"ms,服务：{}"+entry.getValue().id());
            }
        }
    }

}
